package com.data.sample.controller;

import com.data.sample.config.SitesConfig;
import com.data.sample.entity.*;
import com.data.sample.service.IEquipmentUnitService;
import com.data.sample.service.IOpcUaCkMappingService;
import com.data.sample.service.IOpcUaPointService;
import com.data.sample.service.IOpcUaProtocolService;
import com.data.sample.service.impl.KafkaTopicService;
import com.data.sample.service.impl.LinkedBlockingQueueService;
import com.data.sample.util.OpcUAClientRunner;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.milo.opcua.sdk.client.api.identity.AnonymousProvider;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component
@Slf4j
public class OpcUAConfigController {

    @Resource
    IEquipmentUnitService iEquipmentUnitService;

    @Resource
    IOpcUaProtocolService iOpcUaProtocolService;

    @Resource
    IOpcUaPointService iOpcUaPointService;

    @Resource
    IOpcUaCkMappingService iOpcUaCkMappingService;

    @Resource
    private KafkaTopicService kafkaTopicService;

    @Resource
    LinkedBlockingQueueService linkedBlockingQueueService;

    @Resource
    OpcUAClientRunner opcUAClientRunner;

    @Autowired
    private SitesConfig sitesConfig;

    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    @PostConstruct
    public void initYardConfig() {
        String[] yards = sitesConfig.getSites().split(",");
        for (String yard : yards){
            initOPCUAConfig(yard.trim());
        }
    }

    public void initOPCUAConfig(String yard) {
        List<OpcClientConfig> opcClientConfigLst = new ArrayList<>();
        List<EquipmentUnit> equipmentUnitLst = iEquipmentUnitService.selectEquipmentUnit(yard);

        for (EquipmentUnit equipmentUnit :
                equipmentUnitLst) {
            log.info(equipmentUnit.getUnitCode());

            List<OpcUaProtocol> opcUaProtocolLst = iOpcUaProtocolService.selectOpcUaProtocol(equipmentUnit.getId());

            for (OpcUaProtocol opcUaProtocol :
                    opcUaProtocolLst) {
                log.info(opcUaProtocol.getId());

                //1.获取mysql
                List<CkMapping> ckMappingLst = iOpcUaCkMappingService.selectStorage(opcUaProtocol.getId());

                // 转换为 Map<String, String>：key=pointName，value=前缀"P"+ckField
                // 转换为 LinkedHashMap，保持和原 List 相同的顺序
                Map<String, String> ckMappingMap = ckMappingLst.stream()
                        .collect(Collectors.toMap(
                                CkMapping::getPointName,  // key：pointName
                                ck -> "P" + ck.getCkField(),  // value：前缀"P"+ckField
                                (existingValue, newValue) -> existingValue,  // 处理重复key（根据需求选择保留策略）
                                LinkedHashMap::new  // 指定使用 LinkedHashMap 保存结果，保留插入顺序
                        ));

                /**
                 * 获取某个机组的点位信息
                 */
                List<OpcUaPoint> opcUaPointLst = iOpcUaPointService.selectOpcUaPoints(opcUaProtocol.getId());
                /**获取标定映射*/
                Map<String, Calibration> CalibrationMap = opcUAClientRunner.createCalibration(opcUaPointLst);
                /**创建点表协议*/
                Map<String, NodeId> nodeIdMap = opcUAClientRunner.getMonitoredNodeIds(opcUaPointLst);

                OpcClientConfig opcClientConfig = new OpcClientConfig(
                        equipmentUnit.getUnitCode(),
                        opcUaProtocol.getId(),
                        opcUaProtocol.getMainServerAddress(),
                        opcUaProtocol.getPrepareServerAddress(),
                        opcUaProtocol.getCertificatePath(),
                        opcUaProtocol.getUserName(),
                        opcUaProtocol.getPassword(),
                        opcUaProtocol.getDataReleaseInterval(),
                        opcUaProtocol.getKeepAliveCount(),
                        opcUaProtocol.getEnableFlag(),
                        opcUaPointLst,
                        SecurityPolicy.None,
                        new AnonymousProvider(),
                        ckMappingMap,
                        true,
                        true,
                        nodeIdMap,
                        CalibrationMap,
                        opcUaProtocol.getDataReleaseInterval()
                );

                opcClientConfigLst.add(opcClientConfig);
            }
        }


        Map<String, LinkedBlockingQueue<Map<String, Object>>> ckLinkedBlockingQueueMap = linkedBlockingQueueService.getCkQueueMap();

        for (OpcClientConfig opcClientConfig:
                opcClientConfigLst) {

            kafkaTopicService.dynamicCreateTopic(opcClientConfig.getUnitCode(), 1 , (short) 1);

            LinkedBlockingQueue<Map<String, Object>> ckQueue = new LinkedBlockingQueue<>();
            ckLinkedBlockingQueueMap.put(opcClientConfig.getUnitCode(), ckQueue);

            /**注：OPCUA使能标志为1，且监视参数个数不为0时，创建客户端*/
            if (1 == opcClientConfig.getEnableFlag()){

                if (!opcClientConfig.getNodeIdMap().isEmpty()){

                    try {
                        opcUAClientRunner.createClient(opcClientConfig.getMainServerAddress(), opcClientConfig);
                    } catch (Throwable t) {
                        log.error("[{}]OPC UA客户端运行错误: {}", opcClientConfig.getMainServerAddress(), t.getMessage());
                    }

                    try {
                        opcUAClientRunner.createClient(opcClientConfig.getPrepareServerAddress(), opcClientConfig);
                    } catch (Throwable t) {
                        log.error("[{}]OPC UA客户端运行错误: {}", opcClientConfig.getPrepareServerAddress(), t.getMessage());
                    }

                    opcUAClientRunner.run(opcClientConfig);

                    scheduler.scheduleAtFixedRate(() -> {
                        try {
                            //1.获取columns
                            List<String> columns = new ArrayList<>();
                            int size = opcClientConfig.getCkMappingMap().size();
                            columns.add("P1");
                            columns.add("P2");
                            columns.add("P3");
                            for (int i = 0; i < size; i++) {
                                columns.add("P" + (i + 4));
                            }
                            saveData2CK(opcClientConfig, columns);

                        } catch (Exception e) {
                            log.error("{}定时任务执行失败", opcClientConfig.getUnitCode(), e);
                        }
                    }, 10000, 20000, TimeUnit.MILLISECONDS);
                }
            }
        }
    }

    /**
     * 存储数据至CK
     * @param opcClientConfig
     * @param columns
     */
    public void saveData2CK(OpcClientConfig opcClientConfig, List<String> columns) {

        LinkedBlockingQueue<Map<String, Object>> linkedBlockingQueue = linkedBlockingQueueService.getCkQueueMap().get(opcClientConfig.getUnitCode());

        // 创建一个集合接收弹出的所有元素
        List<Map<String, Object>> dataToCkLst = new ArrayList<>();
        // 一次性转移所有元素到allElements（队列会被清空）
        int drainedCount = linkedBlockingQueue.drainTo(dataToCkLst);

        if (drainedCount > 0){
            /**
             * 插入CK数据
             * 输入：表名，列名，数据
             * 输出：已经插入完成的CK数据
             * 功能处理：插入CK数据
             */
            log.info("{}发送至ck数据量：{}", opcClientConfig.getUnitCode(), drainedCount);
            iOpcUaCkMappingService.dynamicBatchInsert("UNIT_" + opcClientConfig.getUnitCode(),
                    columns,
                    dataToCkLst
            );
        }
    }
}

